from __future__ import annotations

import hashlib
import importlib.util
import io
import os
from pathlib import Path
import re
import shutil
import subprocess
import sys
import tempfile
import unittest
from unittest import mock
import zlib

import numpy as np
from PIL import Image, ImageDraw


MODULE_PATH = Path(__file__).resolve().parents[1] / "extract_ppt_slides.py"
SPEC = importlib.util.spec_from_file_location("extract_ppt_slides", MODULE_PATH)
assert SPEC and SPEC.loader
slides = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = slides
SPEC.loader.exec_module(slides)


def feature(index: int, array: np.ndarray, score: float = 0.2, sharpness: float | None = None):
    value = int(np.mean(array))
    phash = int.from_bytes(hashlib.sha256(array.tobytes()).digest()[:8], "big")
    return slides.FrameFeature(
        sample_index=index,
        relative_seconds=float(index),
        gray=array.astype(np.uint8).copy(),
        perceptual_hash=phash,
        content_score=score,
        sharpness=float(value if sharpness is None else sharpness),
    )


def candidate(timestamp: float, array: np.ndarray, score: float, phash: int = 0):
    return slides.PageCandidate(timestamp, array.astype(np.uint8).copy(), phash, score, 1.0)


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest().upper()


def pdf_rgb_pages(path: Path):
    data = path.read_bytes()
    objects = {
        int(match.group(1)): match.group(2)
        for match in re.finditer(rb"(?m)^(\d+) 0 obj\n(.*?)\nendobj\n", data, re.DOTALL)
    }
    pages_body = objects[2]
    kids_match = re.search(rb"/Kids \[(.*?)\]", pages_body, re.DOTALL)
    assert kids_match
    page_ids = [int(value) for value in re.findall(rb"(\d+) 0 R", kids_match.group(1))]
    result = []
    for page_id in page_ids:
        page = objects[page_id]
        media = re.search(rb"/MediaBox \[0 0 (\d+) (\d+)\]", page)
        image_ref = re.search(rb"/Im0 (\d+) 0 R", page)
        assert media and image_ref
        image_body = objects[int(image_ref.group(1))]
        length = int(re.search(rb"/Length (\d+)", image_body).group(1))
        marker = b"stream\n"
        start = image_body.index(marker) + len(marker)
        compressed = image_body[start : start + length]
        result.append(((int(media.group(1)), int(media.group(2))), zlib.decompress(compressed)))
    return result


class ShortReadStream:
    def __init__(self, chunks):
        self.chunks = list(chunks)

    def read(self, size):
        if not self.chunks:
            return b""
        chunk = self.chunks.pop(0)
        if len(chunk) <= size:
            return chunk
        self.chunks.insert(0, chunk[size:])
        return chunk[:size]


class ReadExactTests(unittest.TestCase):
    def test_short_reads_clean_eof_and_partial_eof(self):
        self.assertEqual(slides.read_exact(ShortReadStream([b"a", b"bc", b"def"]), 6), b"abcdef")
        self.assertIsNone(slides.read_exact(ShortReadStream([]), 6))
        with self.assertRaisesRegex(slides.SlideExtractionError, "半帧 EOF"):
            slides.read_exact(ShortReadStream([b"abc"]), 6)


class SegmentAndClassifierTests(unittest.TestCase):
    def test_anchor_drift_emits_only_old_and_new_plateaus(self):
        detector = slides.StableSegmentDetector()
        emitted = []
        arrays = [np.full((8, 8), 20, np.uint8)] * 3
        arrays += [np.full((8, 8), value, np.uint8) for value in (22, 24, 26, 28, 30, 32)]
        arrays += [np.full((8, 8), 80, np.uint8)] * 3
        for index, array in enumerate(arrays):
            value = detector.add(feature(index, array, score=index / 100, sharpness=index))
            if value:
                emitted.append(value)
        final = detector.flush()
        if final:
            emitted.append(final)
        self.assertEqual([round(float(np.mean(item.feature))) for item in emitted], [20, 80])

    def test_animation_increase_decrease_and_equal_score(self):
        classifier = slides.PageClassifier()
        base = np.zeros((8, 8), np.uint8)
        added = base.copy()
        added[:, :2] = 20
        self.assertEqual(classifier.consume(candidate(1, base, 0.20)), "new_page")
        self.assertEqual(classifier.consume(candidate(2, added, 0.30)), "animation_updated")
        self.assertEqual(classifier.pages[0].current_timestamp, 2)
        reduced = base.copy()
        self.assertEqual(classifier.consume(candidate(3, reduced, 0.10)), "animation_retained")
        self.assertEqual(classifier.pages[0].current_timestamp, 2)
        self.assertEqual(
            classifier.consume(candidate(4, added, 0.30 + slides.CONTENT_SCORE_TOLERANCE / 2)),
            "animation_updated",
        )
        self.assertEqual(classifier.pages[0].current_timestamp, 4)

    def test_a_b_a_keeps_first_order(self):
        classifier = slides.PageClassifier()
        a = np.zeros((8, 8), np.uint8)
        b = np.full((8, 8), 255, np.uint8)
        classifier.consume(candidate(1, a, 0.2, 0))
        classifier.consume(candidate(2, b, 0.2, (1 << 64) - 1))
        action = classifier.consume(candidate(3, a, 0.2, 0))
        self.assertEqual(action, "historical_duplicate")
        self.assertEqual(len(classifier.pages), 2)
        self.assertEqual([page.first_seen_order for page in classifier.pages], [0, 1])

    def test_recent_animation_has_priority_over_historical_duplicate(self):
        classifier = slides.PageClassifier()
        historical = np.zeros((8, 8), np.uint8)
        recent = np.full((8, 8), 4, np.uint8)
        classifier.consume(candidate(1, historical, 0.20, 0))
        classifier.pages.append(
            slides.LogicalPage(recent.copy(), recent.copy(), 0, 2.0, 0.20, 1)
        )
        action = classifier.consume(candidate(3, historical, 0.22, 0))
        self.assertEqual(action, "animation_updated")
        self.assertEqual(len(classifier.pages), 2)


class ProcessAndCommandTests(unittest.TestCase):
    def test_command_freezes_time_mapping(self):
        scan = slides._scan_command("ffmpeg", Path("x.mp4"), 320, 180)
        self.assertIn("setpts=PTS-STARTPTS,fps=fps=1:start_time=0,scale=320:180:flags=bilinear", scan)
        extract = slides._extract_command("ffmpeg", Path("x.mp4"), 12.0, Path("x.png"))
        self.assertEqual(extract[extract.index("-ss") + 1], "12.000")
        self.assertEqual(feature(9, np.zeros((2, 2), np.uint8)).relative_seconds, 9.0)

    @unittest.skipUnless(shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg is required")
    def test_vfr_nonzero_pts_scan_and_exact_seek_share_relative_time(self):
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            video = root / "vfr-nonzero.mkv"
            subprocess.run(
                [
                    shutil.which("ffmpeg"), "-hide_banner", "-loglevel", "error",
                    "-f", "lavfi", "-i", "color=red:size=64x64:rate=2:duration=2",
                    "-f", "lavfi", "-i", "color=green:size=64x64:rate=3:duration=2",
                    "-f", "lavfi", "-i", "color=blue:size=64x64:rate=4:duration=2",
                    "-filter_complex", "[0:v][1:v][2:v]concat=n=3:v=1:a=0,setpts=PTS+5/TB[v]",
                    "-map", "[v]", "-fps_mode", "vfr", "-c:v", "ffv1", str(video),
                ],
                check=True,
            )
            start_time = float(
                subprocess.check_output(
                    [
                        shutil.which("ffprobe"), "-v", "error", "-select_streams", "v:0",
                        "-show_entries", "stream=start_time", "-of", "default=nw=1:nk=1", str(video),
                    ],
                    text=True,
                ).strip()
            )
            self.assertGreaterEqual(start_time, 5.0)

            scan_process = subprocess.Popen(
                slides._scan_command(shutil.which("ffmpeg"), video, 64, 64),
                stdin=subprocess.DEVNULL,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            assert scan_process.stdout is not None
            scanned = []
            while True:
                payload = slides.read_exact(scan_process.stdout, 64 * 64 * 3)
                if payload is None:
                    break
                scanned.append(np.frombuffer(payload, np.uint8).reshape((64, 64, 3)))
            scan_process.stdout.close()
            stderr = scan_process.stderr.read() if scan_process.stderr else b""
            if scan_process.stderr:
                scan_process.stderr.close()
            self.assertEqual(scan_process.wait(), 0, stderr.decode(errors="replace"))
            self.assertGreaterEqual(len(scanned), 5)

            extracted = []
            for timestamp in (0.0, 2.0, 4.0):
                path = root / f"at-{timestamp:.0f}.png"
                slides._run_file_command(
                    slides._extract_command(shutil.which("ffmpeg"), video, timestamp, path),
                    root / f"at-{timestamp:.0f}.stderr.log",
                    "VFR 定位",
                )
                with Image.open(path) as image:
                    extracted.append(np.asarray(image.convert("RGB"), np.uint8))

            for sequence in ((scanned[0], scanned[2], scanned[4]), tuple(extracted)):
                means = [np.mean(item, axis=(0, 1)) for item in sequence]
                self.assertEqual([int(np.argmax(value)) for value in means], [0, 1, 2])

    @unittest.skipUnless(shutil.which("ffmpeg"), "FFmpeg is required")
    def test_large_stderr_does_not_deadlock_and_nonzero_is_reported(self):
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            ok_log = root / "ok.log"
            slides._run_file_command(
                [sys.executable, "-c", "import sys; sys.stderr.write('x'*300000)"],
                ok_log,
                "桩命令",
            )
            self.assertEqual(ok_log.stat().st_size, 300000)
            bad_log = root / "bad.log"
            with self.assertRaisesRegex(slides.SlideExtractionError, "exit=7"):
                slides._run_file_command(
                    [sys.executable, "-c", "import sys; sys.stderr.write('boom'); raise SystemExit(7)"],
                    bad_log,
                    "桩命令",
                )

    def test_keyboard_interrupt_reaps_file_process_and_preserves_type(self):
        events = []
        handles = []

        class FakeProcess:
            stdout = None

            def __init__(self):
                self.wait_calls = 0
                self.running = True

            def wait(self, timeout=None):
                self.wait_calls += 1
                events.append(("wait", timeout))
                if self.wait_calls == 1:
                    raise KeyboardInterrupt()
                self.running = False
                return 0

            def poll(self):
                return None if self.running else 0

            def terminate(self):
                events.append(("terminate", None))

            def kill(self):
                events.append(("kill", None))

        def make_process(*args, **kwargs):
            handles.append(kwargs["stderr"])
            return FakeProcess()

        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            source = root / "source.mp4"
            source.write_bytes(b"stub")
            output = root / "result"
            page_feature = np.zeros((8, 8), np.uint8)
            logical_page = slides.LogicalPage(page_feature, page_feature, 0, 1.0, 0.2, 0)
            with mock.patch.object(slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)), mock.patch.object(
                slides, "scan_video", return_value=[logical_page]
            ), mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object(
                slides.subprocess, "Popen", side_effect=make_process
            ):
                with self.assertRaises(KeyboardInterrupt):
                    slides.extract_slides(str(source), str(output))
            self.assertFalse(output.exists())
            self.assertFalse(any(root.glob(".result.staging-*")))
            self.assertTrue(handles[0].closed)
        self.assertEqual(events[1][0], "terminate")
        self.assertEqual(events[2][0], "wait")

    def test_keyboard_interrupt_reaps_scan_process_and_preserves_type(self):
        events = []

        class Stdout:
            def read(self, size):
                raise KeyboardInterrupt()

            def close(self):
                events.append("close")

        class FakeProcess:
            def __init__(self):
                self.stdout = Stdout()
                self.running = True

            def wait(self, timeout=None):
                events.append("wait")
                self.running = False
                return 0

            def poll(self):
                return None if self.running else 0

            def terminate(self):
                events.append("terminate")

            def kill(self):
                events.append("kill")

        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            source = root / "source.mp4"
            source.write_bytes(b"stub")
            output = root / "result"
            with mock.patch.object(
                slides.subprocess, "Popen", return_value=FakeProcess()
            ), mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object(
                slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)
            ):
                with self.assertRaises(KeyboardInterrupt):
                    slides.extract_slides(str(source), str(output))
            self.assertFalse(output.exists())
            self.assertFalse(any(root.glob(".result.staging-*")))
        self.assertEqual(events[:3], ["close", "terminate", "wait"])

    def test_stderr_open_failure_is_wrapped_with_stage(self):
        with tempfile.TemporaryDirectory() as raw, mock.patch.object(
            slides.Path, "open", side_effect=OSError("denied")
        ):
            with self.assertRaisesRegex(slides.SlideExtractionError, "桩阶段失败：denied"):
                slides._run_file_command(["fake"], Path(raw) / "stderr.log", "桩阶段")

    def test_cleanup_error_does_not_replace_keyboard_interrupt(self):
        class FakeProcess:
            stdout = None

            def wait(self, timeout=None):
                raise KeyboardInterrupt()

            def poll(self):
                return None

            def terminate(self):
                raise OSError("cleanup denied")

        with tempfile.TemporaryDirectory() as raw, mock.patch.object(
            slides.subprocess, "Popen", return_value=FakeProcess()
        ):
            with self.assertRaises(KeyboardInterrupt):
                slides._run_file_command(["fake"], Path(raw) / "stderr.log", "桩阶段")

    def test_stderr_close_error_preserves_interrupt_in_both_helpers(self):
        class CloseFailingHandle:
            def close(self):
                raise OSError("stderr close denied")

            def flush(self):
                return None

        class FileProcess:
            stdout = None

            def __init__(self):
                self.calls = 0

            def wait(self, timeout=None):
                self.calls += 1
                if self.calls == 1:
                    raise KeyboardInterrupt()
                return 0

            def poll(self):
                return None

            def terminate(self):
                return None

        class ScanStdout:
            def read(self, size):
                raise KeyboardInterrupt()

            def close(self):
                return None

        class ScanProcess:
            def __init__(self):
                self.stdout = ScanStdout()

            def wait(self, timeout=None):
                return 0

            def poll(self):
                return None

            def terminate(self):
                return None

        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object(
                slides.subprocess, "Popen", return_value=FileProcess()
            ):
                with self.assertRaises(KeyboardInterrupt):
                    slides._run_file_command(["fake"], root / "file.stderr.log", "文件阶段")
            with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object(
                slides.subprocess, "Popen", return_value=ScanProcess()
            ), mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"):
                with self.assertRaises(KeyboardInterrupt):
                    slides.scan_video(Path("x.mp4"), root, slides.VideoInfo(640, 360, 10))

    def test_stderr_close_error_is_staged_on_normal_and_ordinary_failure(self):
        class CloseFailingHandle:
            def close(self):
                raise OSError("stderr close denied")

            def flush(self):
                return None

        class FakeProcess:
            stdout = None

            def __init__(self, code):
                self.code = code

            def wait(self, timeout=None):
                return self.code

            def poll(self):
                return self.code

        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object(
                slides.subprocess, "Popen", return_value=FakeProcess(0)
            ):
                with self.assertRaisesRegex(slides.SlideExtractionError, "文件阶段关闭 stderr 失败"):
                    slides._run_file_command(["fake"], root / "normal.log", "文件阶段")
            with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object(
                slides.subprocess, "Popen", return_value=FakeProcess(7)
            ), mock.patch.object(slides, "_tail", return_value="boom"):
                with self.assertRaisesRegex(slides.SlideExtractionError, "文件阶段失败（exit=7）"):
                    slides._run_file_command(["fake"], root / "failed.log", "文件阶段")


class PdfTests(unittest.TestCase):
    def test_pdf_rgb_content_and_order(self):
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            paths = []
            for index, color in enumerate(((255, 20, 20), (20, 40, 255))):
                path = root / f"{index}.png"
                Image.new("RGB", (17 + index, 11 + index), color).save(path)
                paths.append(path)
            pdf = root / "slides.pdf"
            slides.write_image_pdf(paths, pdf)
            decoded = pdf_rgb_pages(pdf)
            self.assertEqual([item[0] for item in decoded], [(17, 11), (18, 12)])
            for path, (_, rgb) in zip(paths, decoded):
                with Image.open(path) as image:
                    self.assertEqual(hashlib.sha256(rgb).digest(), hashlib.sha256(image.convert("RGB").tobytes()).digest())
            pdfinfo = shutil.which("pdfinfo")
            if pdfinfo and Path(pdfinfo).suffix.lower() not in {".cmd", ".bat"}:
                result = subprocess.run([pdfinfo, str(pdf)], capture_output=True, text=True, check=True)
                self.assertIn("Pages:           2", result.stdout)

    def test_twenty_pages_keep_one_decoded_image_active(self):
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            paths = []
            for index in range(20):
                path = root / f"{index:02d}.png"
                Image.new("RGB", (12, 8), (index, index * 2, index * 3)).save(path)
                paths.append(path)
            active = 0
            maximum = 0

            class CountedOpen:
                def __init__(self, path):
                    self.image = Image.open(path)

                def __enter__(self):
                    nonlocal active, maximum
                    active += 1
                    maximum = max(maximum, active)
                    return self.image

                def __exit__(self, exc_type, exc, traceback):
                    nonlocal active
                    self.image.close()
                    active -= 1

            pdf = root / "twenty.pdf"
            slides.write_image_pdf(paths, pdf, image_opener=CountedOpen)
            self.assertEqual(maximum, 1)
            self.assertEqual(active, 0)
            self.assertEqual(len(pdf_rgb_pages(pdf)), 20)


class SafetyAndIntegrationTests(unittest.TestCase):
    def test_existing_output_is_rejected_before_external_process(self):
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            source = root / "source.mp4"
            source.write_bytes(b"not a video")
            for kind in ("directory", "file"):
                with self.subTest(kind=kind):
                    output = root / f"existing-{kind}"
                    if kind == "directory":
                        output.mkdir()
                    else:
                        output.write_bytes(b"occupied")
                    with mock.patch.object(slides.subprocess, "Popen") as popen:
                        with self.assertRaisesRegex(slides.SlideExtractionError, "默认不覆盖"):
                            slides.extract_slides(str(source), str(output))
                    popen.assert_not_called()

    def test_live_and_broken_output_links_are_rejected_by_lexical_precheck(self):
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            source = root / "source.mp4"
            source.write_bytes(b"stub")
            for link_kind in ("live", "broken"):
                with self.subTest(link_kind=link_kind):
                    output = slides._absolute_lexical(root / f"{link_kind}-link")
                    original_lexists = slides._lexists

                    def controlled_lexists(path):
                        return True if path == output else original_lexists(path)

                    with mock.patch.object(slides, "_lexists", side_effect=controlled_lexists), mock.patch.object(
                        slides.subprocess, "Popen"
                    ) as popen:
                        with self.assertRaisesRegex(slides.SlideExtractionError, "默认不覆盖"):
                            slides.extract_slides(str(source), str(output))
                    popen.assert_not_called()

    def test_input_symlink_is_rejected_before_external_process(self):
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            source = slides._absolute_lexical(root / "source-link.mp4")
            with mock.patch.object(slides, "_is_symlink", side_effect=lambda path: path == source), mock.patch.object(
                slides.subprocess, "Popen"
            ) as popen:
                with self.assertRaisesRegex(slides.SlideExtractionError, "拒绝符号链接"):
                    slides.extract_slides(str(source), str(root / "result"))
            popen.assert_not_called()

    def test_baseexception_removes_staging_and_leaves_no_formal_output(self):
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            source = root / "source.mp4"
            source.write_bytes(b"stub")
            output = root / "result"
            with mock.patch.object(slides, "probe_video", side_effect=KeyboardInterrupt()):
                with self.assertRaises(KeyboardInterrupt):
                    slides.extract_slides(str(source), str(output))
            self.assertFalse(output.exists())
            self.assertFalse(any(root.glob(".result.staging-*")))

    def test_commit_conflict_is_wrapped_and_staging_is_removed(self):
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            source = root / "source.mp4"
            source.write_bytes(b"stub")
            output = root / "result"
            with mock.patch.object(slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)), mock.patch.object(
                slides, "scan_video", return_value=[]
            ), mock.patch.object(slides, "extract_original_pages", return_value=[]), mock.patch.object(
                slides, "write_image_pdf"
            ), mock.patch.object(slides.os, "rename", side_effect=FileExistsError("race")):
                with self.assertRaisesRegex(slides.SlideExtractionError, "提交正式输出目录失败：race"):
                    slides.extract_slides(str(source), str(output))
            self.assertFalse(output.exists())
            self.assertFalse(any(root.glob(".result.staging-*")))

    @unittest.skipUnless(shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg is required")
    def test_chinese_path_synthetic_video_end_to_end(self):
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw) / "中文路径"
            root.mkdir()
            frame_root = root / "frames"
            frame_root.mkdir()
            size = (640, 360)

            def page_a(complete: bool):
                image = Image.new("RGB", size, "white")
                draw = ImageDraw.Draw(image)
                draw.rectangle((0, 0, 639, 65), fill=(26, 78, 145))
                draw.rectangle((55, 100, 585, 125), fill=(30, 30, 30))
                if complete:
                    draw.rectangle((55, 165, 470, 190), fill=(30, 30, 30))
                    draw.rectangle((55, 230, 540, 255), fill=(30, 30, 30))
                    draw.rectangle((570, 320, 580, 330), fill=(220, 20, 20))
                return image

            def page_b():
                image = Image.new("RGB", size, (18, 34, 68))
                draw = ImageDraw.Draw(image)
                draw.rectangle((65, 75, 575, 285), fill=(230, 190, 45))
                draw.rectangle((110, 125, 530, 155), fill=(18, 34, 68))
                return image

            frames = [page_a(False)] * 3
            frames += [Image.new("RGB", size, (127, 127, 127))]
            frames += [page_a(True)] * 3
            frames += [Image.new("RGB", size, (90, 20, 130))]
            frames += [page_b()] * 3
            frames += [Image.new("RGB", size, (20, 150, 90))]
            frames += [page_a(True)] * 3
            for index, image in enumerate(frames):
                image.save(frame_root / f"frame_{index:03d}.png")

            source = root / "合成会议录屏.mp4"
            subprocess.run(
                [
                    shutil.which("ffmpeg"), "-hide_banner", "-loglevel", "error", "-framerate", "1",
                    "-i", str(frame_root / "frame_%03d.png"), "-c:v", "libx264rgb", "-crf", "0", "-pix_fmt", "rgb24", str(source),
                ],
                check=True,
            )
            before = (source.stat().st_size, source.stat().st_ctime_ns, source.stat().st_mtime_ns, sha256(source))
            output = root / "验收输出"
            result = subprocess.run(
                [sys.executable, str(MODULE_PATH), str(source), "--output", str(output)],
                capture_output=True,
                text=True,
                timeout=120,
            )
            self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
            self.assertEqual(
                {entry.name for entry in output.iterdir()},
                {"pages", "合成会议录屏.slides.pdf"},
            )
            pngs = sorted((output / "pages").glob("slide_*.png"))
            self.assertEqual(len(pngs), 2, result.stdout)
            for png in pngs:
                with Image.open(png) as image:
                    self.assertEqual(image.size, size)
            expected = [page_a(True), page_b()]
            for actual_path, expected_image in zip(pngs, expected):
                with Image.open(actual_path) as actual:
                    self.assertEqual(
                        hashlib.sha256(actual.convert("RGB").tobytes()).digest(),
                        hashlib.sha256(expected_image.tobytes()).digest(),
                    )
            decoded = pdf_rgb_pages(output / "合成会议录屏.slides.pdf")
            self.assertEqual(len(decoded), 2)
            for png, (_, rgb) in zip(pngs, decoded):
                with Image.open(png) as image:
                    self.assertEqual(hashlib.sha256(rgb).digest(), hashlib.sha256(image.convert("RGB").tobytes()).digest())
            after = (source.stat().st_size, source.stat().st_ctime_ns, source.stat().st_mtime_ns, sha256(source))
            self.assertEqual(after, before)
            self.assertFalse(any(root.glob(".验收输出.staging-*")))


if __name__ == "__main__":
    unittest.main()