video-image-production-build/tests/run/test_pipeline_entry.py

from __future__ import annotations

import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from production_build_agents.run.layout import (
    global_data_delivery_path,
    run_directory,
)
from run_pipeline import _build_parser, main, run_pipeline


class PipelineEntryTest(unittest.TestCase):
    def _inputs(self, root: Path) -> tuple[Path, Path, Path]:
        source = root / "production_final.json"
        source.write_text('{"fixture":true}', encoding="utf-8")
        anchor = root / "anchor.png"
        anchor.write_bytes(b"anchor")
        return source, anchor, root / "runs"

    def _completed_global_result(
        self,
        output_root: Path,
        *,
        thread_id: str,
    ) -> dict[str, object]:
        delivery = global_data_delivery_path(
            run_directory(output_root, thread_id)
        ).resolve()
        delivery.parent.mkdir(parents=True, exist_ok=True)
        delivery.write_text('{"schema_version":"0.3"}', encoding="utf-8")
        return {
            "run_id": thread_id,
            "status": "COMPLETED",
            "phase": "FINALIZE",
            "global_data_delivery_path": str(delivery),
        }

    def test_completed_global_data_starts_production_in_order(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            source, anchor, output_root = self._inputs(root)
            calls: list[str] = []

            def global_runner(_source, **kwargs):
                calls.append("global_data")
                self.assertEqual(kwargs["thread_id"], "v24-global-data")
                self.assertEqual(kwargs["pipeline_round_id"], "v24")
                return self._completed_global_result(
                    output_root,
                    thread_id=kwargs["thread_id"],
                )

            def production_runner(delivery, **kwargs):
                calls.append("production")
                self.assertEqual(
                    delivery,
                    global_data_delivery_path(
                        run_directory(output_root, "v24-global-data")
                    ).resolve(),
                )
                self.assertEqual(
                    kwargs["thread_id"],
                    "v24-production",
                )
                self.assertEqual(
                    kwargs["shared_visual_anchor_path"],
                    anchor,
                )
                self.assertEqual(kwargs["max_plan_revisions"], 2)
                self.assertTrue(
                    kwargs["pause_after_progress_review"]
                )
                self.assertEqual(kwargs["pipeline_round_id"], "v24")
                self.assertEqual(
                    kwargs["upstream_global_data_thread_id"],
                    "v24-global-data",
                )
                return {
                    "run_id": "v24-production",
                    "status": "RUNNING",
                    "phase": "PREPARE_SEGMENT",
                }

            with (
                patch(
                    "run_pipeline.run_global_data",
                    side_effect=global_runner,
                ),
                patch(
                    "run_pipeline.run_full_production",
                    side_effect=production_runner,
                ),
            ):
                result = run_pipeline(
                    source,
                    output_root=output_root,
                    round_id="v24",
                    shared_visual_anchor_path=anchor,
                    max_plan_revisions=2,
                    pause_after_progress_review=True,
                )

        self.assertEqual(calls, ["global_data", "production"])
        self.assertEqual(result["status"], "RUNNING")
        self.assertEqual(result["phase"], "PRODUCTION")
        self.assertEqual(
            result["global_data_thread_id"],
            "v24-global-data",
        )
        self.assertEqual(
            result["production_thread_id"],
            "v24-production",
        )

    def test_noncompleted_global_data_never_starts_production(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            source, anchor, output_root = self._inputs(root)
            with (
                patch(
                    "run_pipeline.run_global_data",
                    return_value={
                        "run_id": "v24-global-data",
                        "status": "FAILED",
                        "phase": "VALIDATE_GLOBAL_DATA",
                    },
                ),
                patch(
                    "run_pipeline.run_full_production"
                ) as production,
            ):
                result = run_pipeline(
                    source,
                    output_root=output_root,
                    round_id="v24",
                    shared_visual_anchor_path=anchor,
                )

        production.assert_not_called()
        self.assertEqual(result["status"], "FAILED")
        self.assertEqual(result["phase"], "GLOBAL_DATA")
        self.assertIsNone(result["production_result"])

    def test_repeated_round_reuses_both_stable_run_identities(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            source, anchor, output_root = self._inputs(root)
            global_ids: list[str] = []
            production_ids: list[str] = []
            production_statuses = iter(("RUNNING", "COMPLETED"))

            def global_runner(_source, **kwargs):
                global_ids.append(kwargs["thread_id"])
                return self._completed_global_result(
                    output_root,
                    thread_id=kwargs["thread_id"],
                )

            def production_runner(_delivery, **kwargs):
                production_ids.append(kwargs["thread_id"])
                status = next(production_statuses)
                return {
                    "run_id": kwargs["thread_id"],
                    "status": status,
                    "phase": (
                        "PREPARE_SEGMENT"
                        if status == "RUNNING"
                        else "FINALIZE"
                    ),
                }

            with (
                patch(
                    "run_pipeline.run_global_data",
                    side_effect=global_runner,
                ),
                patch(
                    "run_pipeline.run_full_production",
                    side_effect=production_runner,
                ),
            ):
                first = run_pipeline(
                    source,
                    output_root=output_root,
                    round_id="v24",
                    shared_visual_anchor_path=anchor,
                    pause_after_progress_review=True,
                )
                second = run_pipeline(
                    source,
                    output_root=output_root,
                    round_id="v24",
                    shared_visual_anchor_path=anchor,
                )

        self.assertEqual(first["status"], "RUNNING")
        self.assertEqual(second["status"], "COMPLETED")
        self.assertEqual(
            global_ids,
            ["v24-global-data", "v24-global-data"],
        )
        self.assertEqual(
            production_ids,
            ["v24-production", "v24-production"],
        )

    def test_completed_result_must_point_to_its_own_delivery(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            source, anchor, output_root = self._inputs(root)
            foreign = root / "foreign-delivery.json"
            foreign.write_text("{}", encoding="utf-8")
            with (
                patch(
                    "run_pipeline.run_global_data",
                    return_value={
                        "run_id": "v24-global-data",
                        "status": "COMPLETED",
                        "global_data_delivery_path": str(foreign),
                    },
                ),
                patch(
                    "run_pipeline.run_full_production"
                ) as production,
                self.assertRaisesRegex(
                    ValueError,
                    "非当前 Run",
                ),
            ):
                run_pipeline(
                    source,
                    output_root=output_root,
                    round_id="v24",
                    shared_visual_anchor_path=anchor,
                )

        production.assert_not_called()

    def test_completed_result_requires_existing_delivery(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            source, anchor, output_root = self._inputs(root)
            expected = global_data_delivery_path(
                run_directory(output_root, "v24-global-data")
            ).resolve()
            with (
                patch(
                    "run_pipeline.run_global_data",
                    return_value={
                        "run_id": "v24-global-data",
                        "status": "COMPLETED",
                        "global_data_delivery_path": str(expected),
                    },
                ),
                patch(
                    "run_pipeline.run_full_production"
                ) as production,
                self.assertRaises(FileNotFoundError),
            ):
                run_pipeline(
                    source,
                    output_root=output_root,
                    round_id="v24",
                    shared_visual_anchor_path=anchor,
                )

        production.assert_not_called()

    def test_round_id_rejects_path_traversal_before_any_run(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            source, anchor, output_root = self._inputs(root)
            with (
                patch("run_pipeline.run_global_data") as global_runner,
                self.assertRaisesRegex(ValueError, "round_id"),
            ):
                run_pipeline(
                    source,
                    output_root=output_root,
                    round_id="../outside",
                    shared_visual_anchor_path=anchor,
                )

        global_runner.assert_not_called()

    def test_cli_requires_stable_round_and_anchor(self) -> None:
        parser = _build_parser()
        args = parser.parse_args(
            [
                "production_final.json",
                "--round-id",
                "v24",
                "--shared-visual-anchor",
                "anchor.png",
                "--pause-after-progress-review",
            ]
        )
        self.assertEqual(args.round_id, "v24")
        self.assertTrue(args.pause_after_progress_review)

    def test_cli_treats_requested_production_pause_as_success(self) -> None:
        argv = [
            "run_pipeline.py",
            "production_final.json",
            "--round-id",
            "v24",
            "--shared-visual-anchor",
            "anchor.png",
            "--pause-after-progress-review",
        ]
        with (
            patch("sys.argv", argv),
            patch(
                "run_pipeline.run_pipeline",
                return_value={
                    "round_id": "v24",
                    "status": "RUNNING",
                    "phase": "PRODUCTION",
                },
            ),
        ):
            main()

    def test_cli_fails_when_global_data_does_not_complete(self) -> None:
        argv = [
            "run_pipeline.py",
            "production_final.json",
            "--round-id",
            "v24",
            "--shared-visual-anchor",
            "anchor.png",
        ]
        with (
            patch("sys.argv", argv),
            patch(
                "run_pipeline.run_pipeline",
                return_value={
                    "round_id": "v24",
                    "status": "FAILED",
                    "phase": "GLOBAL_DATA",
                },
            ),
            self.assertRaises(SystemExit) as raised,
        ):
            main()

        self.assertEqual(raised.exception.code, 1)


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